package com.haisen.mr.tq;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 天气mapreduce算法
 * 找出每个月气温最高的两天：以yyyyMM为key，分组，然后求最高两天
 */
public class TQMR {
	
	
	public static void main(String[] args) throws Exception {
		
		
		
		//1,conf
		Configuration conf = new Configuration(true);
		
		//2,job
		Job job = Job.getInstance(conf);
		job.setJarByClass(TQMR.class);
		
		//3,input,output
		
		Path input = new Path("/user/tq/input");
		FileInputFormat.addInputPath(job, input);
		
		Path output = new Path("/user/tq/output");
		if(output.getFileSystem(conf).exists(output)){
			output.getFileSystem(conf).delete(output,true);
		}
		FileOutputFormat.setOutputPath(job, output );
		
		
		//4,map
		job.setMapperClass(TqMapper.class);
		//key比较器，相同的key一组
		job.setMapOutputKeyClass(TQ.class);
		job.setMapOutputValueClass(Text.class);




		//5,reduce
		job.setReducerClass(TqReducer.class);
		job.setNumReduceTasks(2);
		
		//6,other:sort,part..,group...
		job.setPartitionerClass(TqPartitioner.class);
		//在相同的年月日(key)下对温度降序排列
		job.setSortComparatorClass(TqSortComparator.class);
		job.setGroupingComparatorClass(TqGroupingComparator.class);

       /* 如果配置了setCombinerClass（模板里配置了的），正确的数据流：mapper(k1,v1,k2,v2)->combine(k2,v2,k3,v3)->reducer(k3,v3,k4,v4)。但是通常把combineClass设置为reducer，这样，实际的数据流：mapper(k1,v1,k2,v2)->combine(k2,v2,k3,v3)->reducer(k2,v2,k3,v3)。肯定会出错呀。*/

         //job.setCombinerClass(TqReducer.class);
        // job.setCombinerKeyGroupingComparatorClass(TqGroupingComparator.class);

		job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
		//7,submit
		
		job.waitForCompletion(true);

		
		
		
		
		
		
		
	}
	

}
